package com.vodafone.mapreduce.map;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

import com.vodafone.mapreduce.MapReduce;
import com.vodafone.mapreduce.dto.common.KeyValuePair;
import com.vodafone.mapreduce.dto.function.CombineInput;

/**
 * TODO: Insert your comments here!
 */
public abstract class CombineFunction<K, V, L, W> {

    public static Logger log = MapReduce.LOG;

    protected CombineInput<K, V, L, W> mapResult;
    protected final Map<L, List<W>> intermediate = new HashMap<L, List<W>>();
    protected final AtomicBoolean ready = new AtomicBoolean(false);

    public void initialize(CombineInput<K, V, L, W> mapResult) {
        ready.set(false);
        intermediate.clear();
        this.mapResult = mapResult;
    }

    public void combine() {
        List<KeyValuePair<L, W>> uncombined = mapResult.getIntermediate();
        for (KeyValuePair<L, W> kvp : uncombined) {
            if (intermediate.containsKey(kvp.getKey())) {
                // Map entry in the combine intermediate already exists. Value of the current kvp has to be appended or merged.
                doCombine(kvp);
            } else {
                // Add new map entry in the combine intermediate. Simply add the value of the current kvp.
                List<W> list = new ArrayList<W>();
                list.add(kvp.getValue());
                intermediate.put(kvp.getKey(), list);
            }
        }
        ready.set(true);
    }

    protected abstract void doCombine(KeyValuePair<L, W> kvp);

    public Map<L, List<W>> getCombineResult() {
        if (ready.get()) {
            return intermediate;
        } else {
            return null;
        }
    }

}
